<?php
use Swoole\Http\Response;
use Swoole\WebSocket\Server;
use Swoole\Http\Request;

class WebSocketServer
{
    public Server $server;

    public function __construct()
    {
        $this->delKey();
        $this->server = new Server("0.0.0.0", 9501);
        $this->server->on('handshake', function (Request $request, Response $response) {
            $this->checkToken($request, $response);
            // websocket握手连接算法验证
            $secWebSocketKey = $request->header['sec-websocket-key'];
            $patten = '#^[+/0-9A-Za-z]{21}[AQgw]==$#';
            if (0 === preg_match($patten, $secWebSocketKey) || 16 !== strlen(base64_decode($secWebSocketKey))) {
                $response->end();
                return false;
            }
            $key = base64_encode(
                sha1($request->header['sec-websocket-key'] . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11', true)
            );
            $headers = [
                'Upgrade' => 'websocket',
                'Connection' => 'Upgrade',
                'Sec-WebSocket-Accept' => $key,
                'Sec-WebSocket-Version' => '13',
            ];
            if (isset($request->header['sec-websocket-protocol'])) {
                $headers['Sec-WebSocket-Protocol'] = $request->header['sec-websocket-protocol'];
            }
            foreach ($headers as $key => $val) {
                $response->header($key, $val);
            }
            $response->status(101);
            $response->end();
        });

        $this->server->on('open', function (Server $server, $request){
            echo "server: handshake success with fd{$request->fd}\n";
        });

        $this->server->on('message', function (Server $server, $frame) {
            echo "receive from {$frame->fd}:{$frame->data},opcode:{$frame->opcode},fin:{$frame->finish}\n";
            $this->acceptMessage($frame->data,$server,$frame->fd);
        });

        $this->server->on('close', function ($server, $fd) {
            echo "client {$fd} closed\n";
            $this->redis()->del("user_fd_token:".$fd);
        });

        Swoole\Timer::tick(1000, function (int $timer_id, $server)  {
            //如果广播队列有数据，则广播
            $broadcast = $this->redis()->lPop('broadcast');
            if ($broadcast) {
                foreach ($server->connections as $fd) {
                    // 需要先判断是否是正确的websocket连接，否则有可能会push失败
                    if ($server->isEstablished($fd)) {
                        $server->push($fd, "hello word");
                    }
                }
            }
        }, $this->server);
        $this->server->start();
    }

    /**
     * 获取redis实例
     * @return Redis|void
     */
    private function redis()
    {
        $redis = new \Redis();
        try {
            $redis->connect('127.0.0.1', 6379);
            $redis->select(10);
            $redis->setOption(\Redis::OPT_SERIALIZER, \Redis::SERIALIZER_PHP);
            $redis->getOption(\Redis::OPT_SERIALIZER);
        } catch (RedisException $e) {
            exit($e->getMessage());
        }
        return $redis;
    }

    /**
     * 验证用户连接token
     * @param Request $request
     * @param Response $response
     * @return void
     * @throws RedisException
     */
    private function checkToken(Request $request,Response $response): void
    {
        if (!$request->get) {
            $response->end();
            return;
        }
        if (!$request->get['token']) {
            $response->end();
            return;
        }
        $token = $request->get['token'];
        if (!$this->redis()->exists("user_token:".$token)) {
            $response->end();
            return;
        }
        $this->redis()->set("user_fd:".$token,$request->fd);
        $this->redis()->set("user_fd_token:".$request->fd,$token);
    }

    /**
     * 消息处理
     * @param string $data
     * @param Server $server
     * @return void
     * @throws RedisException
     */
    private function acceptMessage(string $data,Server $server,int $fd): void
    {
        $data = json_decode($data,true);
        if ($data['push_type']=='private') {
            $to_fd = $this->redis()->get("user_fd:".$data['to_uid']);
            $message = [
                'object_name'=>$data['object_name'],
                'push_type'=>$data['push_type'],
                'form_uid'=>$data['form_uid'],
                "content"=>$data['content'],
            ];
            if ($server->isEstablished($to_fd)) {
                $server->push($to_fd, json_encode($message));
                $this->redis()->zAdd("user_reply:" . $data['form_uid'], time(), $data['to_uid']);
                $this->redis()->zAdd("user_reply:" . $data['to_uid'], time(), $data['form_uid']);
            }
        }
        if ($data['push_type']=='group') {
            $group_id = $this->redis()->get("user_group:".$data['to_group_id']);
            $group_id = explode(',',$group_id);
            foreach ($group_id as $vo) {
                $to_fd = $this->redis()->get("user_fd:" . $vo);
                $message = [
                    'object_name' => $data['object_name'],
                    'push_type' => $data['push_type'],
                    'form_uid' => $data['form_uid'],
                    "content" => $data['content'],
                ];
                if ($server->isEstablished($to_fd) && $data['form_uid']!=$vo) {
                    $server->push($to_fd, json_encode($message));
                }
            }
        }
        if ($data['push_type']=='ping') {
            $server->push($fd, json_encode(['push_type'=>"pong"]));
        }
    }

    /**
     * 重启websocket服务,删除key
     * @return void
     * @throws RedisException
     */
    private function delKey(): void
    {
        $this->redis()->del($this->redis()->keys("user_fd:*"));
        $this->redis()->del($this->redis()->keys("user_fd_token:*"));
    }
}

new WebSocketServer();
